What problem does this PR solve?

Spark sum(bigint) on HashAgg can spend significant time in per-group scalar updates. This PR adds SumAggregateSparkInt64SubOp as the default aggregate for non-decimal spark_sum(bigint) → bigint on Linux aarch64 with 256-bit SVE (svcntb()==32), providing batch SVE updates when Spark overflow checking is enabled. The fast path supports flat, constant, and dictionary inputs. Other platforms or SVE widths use original SumAggregateBase. Disable SubOp with BOLT_SPARK_SUM_INT64_USE_SUBOP=0.

Issue Number: N/A

Type of Change

• 🐛 Bug fix (non-breaking change which fixes an issue)
• ✨ New feature (non-breaking change which adds functionality)
• 🚀 Performance improvement (optimization)
• ⚠️ Breaking change (fix or feature that would cause existing functionality to change)
• 🔨 Refactoring (no logic changes)
• 🔧 Build/CI or Infrastructure changes
• 📝 Documentation only

Description

What

• Add SumAggregateSparkInt64SubOp for Spark non-decimal sum(bigint) → bigint (default via registerSum unless env opts out).
• Unified updateBatch for addRawInput / addIntermediateResults.
• sumInt64SubOpCanUseSveKernel(): Linux aarch64 + HWCAP_SVE + svcntb() == 32 (256-bit SVE only); evaluated before decode.
• Batch glue updateGroupsFromDecoded → sveHashAggBatchUpdateGroupSums (SVE kernel, aarch64-only TU, -march=armv8-a+sve).
• sveAccumulateFlaggedRows: batch accumulation for flat, constant, and dictionary-encoded values.
• Extend DecodedVector: hashAggNullsLayoutMode / hashAggIndicesLayoutMode (mode1/mode2 — orthogonal null vs value layout) + hashAggMutable* buffer accessors for batch glue.

Disable SubOp (BOLT_SPARK_SUM_INT64_USE_SUBOP)

Executor example (Gluten / Spark): spark.executorEnv.BOLT_SPARK_SUM_INT64_USE_SUBOP=0.

Registration

registerSum (BIGINT, non-decimal)
  └─ sparkSumInt64UseSubOpFromEnv() ?
        ├─ true  (default: unset / empty env) → SumAggregateSparkInt64SubOp
        └─ false (0 / false / no / off)     → SumAggregate<int64_t,…>

Runtime call chain (updateBatch; raw / intermediate symmetric)

1. mayPushdown && top-level Lazy → Base (pushdown, same as SumAggregateBase::updateInternal)
2. Else Overflow? No → Base
3. Else sumInt64SubOpCanUseSveKernel()? No → Base (no decode; covers non-Linux aarch64, no SVE, or non-256-bit SVE VL)
4. Else decode; if decoded input is indirect lazy (e.g. Dictionary(LazyVector)) → load via callable hook, return
5. Else updateGroupsFromDecoded → sveHashAggBatchUpdateGroupSums, return

SVE kernel supports nullable and non-nullable input batches.

flowchart TD
  A[updateBatch] --> B{top-level Lazy+ pushdown?}
  B -->|yes| C[SumAggregateBase]
  B -->|no| D{Overflow?}
  D -->|no| C
  D -->|yes| E{canUseSveKernel?}
  E -->|no| C
  E -->|yes| F[decode]
  F --> G{indirect lazy?base == LAZY}
  G -->|yes| H[SimpleCallableHook+ LazyVector::load]
  G -->|no| I[updateGroupsFromDecoded→ Sve.cpp]

Behavior summary

Source files and SVE path

flowchart LR
  R["SumAggregate.cpp"]
  H["SubOp.h"]
  CPP["SubOp.cpp"]
  DV["DecodedVector"]
  Hook["SimpleCallableHook"]
  SVE["Sve.cpp"]
  Base["SumAggregateBase"]

  R -->|"registerSum"| H
  H --> CPP
  CPP -->|"decode"| DV
  DV -->|"indirect lazy"| Hook
  DV -->|"flat/const/dict"| SVE
  CPP -->|"pre-decode fallback"| Base

  classDef prNew fill:#d4edda,stroke:#28a745,stroke-width:2px,color:#000
  classDef prModified fill:#fff3cd,stroke:#ffc107,stroke-width:2px,color:#000
  classDef existing fill:#f6f8fa,stroke:#6c757d,color:#000
  class H,CPP,SVE prNew
  class R,DV prModified
  class Base,Hook existing

Tests (11 cases added for this PR; names below are gtest suffixes)

Other SumAggregationTest cases in the same binary (e.g. overflow, decimalSum) are pre-existing, not part of this PR.

cmake --build --preset conan-release --target bolt_functions_spark_aggregates_test bolt_vector_test

ctest --test-dir _build/Release -R '^bolt_functions_spark_aggregates_test$' --output-on-failure -V -- \
  --gtest_filter='SumAggregationTest.sumInt64SubOp*'

ctest --test-dir _build/Release -R '^bolt_vector_test$' --output-on-failure -V -- \
  --gtest_filter='DecodedVectorTest.hashAggLayoutModes'

Pass --gtest_filter after -- so ctest runs only SubOp-related cases (without it, the aggregate binary runs all registered tests).

Note: sumInt64SubOpEnvOffParity skips on Windows (POSIX setenv). SVE instruction coverage requires Linux aarch64 with 256-bit SVE (svcntb()==32); otherwise SubOp falls back to Base and tests still pass.

Performance Impact

• No Impact
• Positive Impact: SVE batch path on flat, constant, dictionary, and typical HashAgg batch shapes; Base fallback skips decode when 256-bit SVE is unavailable. Full TPC-DS 1T self-test: results match baseline; suite-wide E2E gain is limited because int64 sum is a small workload share. Benchmark Results (expand below): TPC-DS 1T HashAgg micro-query on store_sales (group-by + multiple sum(bigint) / decimal-unscaled sums; qtest×5) — ~6.3% mean E2E wall-time vs baseline where sum updates dominate.

SELECT
  ss_item_sk AS item_sk,
  d_date AS solddate,
  count(*) AS cnt,
  sum(cast(ss_quantity AS bigint)) AS sum_qty,
  sum(ss_ext_sales_price) AS sum_ext_sales,
  sum(ss_ext_discount_amt) AS sum_ext_discount,
  sum(ss_net_paid) AS sum_net_paid,
  sum(ss_net_paid_inc_tax) AS sum_net_paid_tax,
  sum(ss_net_paid_inc_tax - ss_net_paid) AS sum_tax,
  sum(ss_ext_wholesale_cost) AS sum_cost,
  sum(ss_ext_list_price) AS sum_ext_list_price,
  sum(ss_net_profit) AS sum_profit
FROM store_sales, date_dim
WHERE ss_sold_date_sk = d_date_sk
GROUP BY ss_item_sk, d_date
ORDER BY sum_profit DESC, ss_item_sk, d_date
LIMIT 100;

• Negative Impact

Release Note

Release Note:

Release Note:
- Spark sum(bigint) on HashAgg: AArch64 SVE batch updates via SumAggregateSparkInt64SubOp (default on).
- SVE batch path: flat, constant, and dictionary inputs when overflow checking is enabled.
- Requires Linux aarch64 + 256-bit SVE (svcntb()==32); otherwise SumAggregateBase.
- Disable SubOp (restore `SumAggregateBase`): `spark.executorEnv.BOLT_SPARK_SUM_INT64_USE_SUBOP=0` (or `false` / `no` / `off`).

Checklist (For Author)

• I have added/updated unit tests (ctest).
• I have verified the code with local build (Release/Debug) on Linux aarch64 + SVE (Gluten Bolt bundle; see Benchmark Results).
• I have run clang-format / linters.
• (Optional) I have run Sanitizers (ASAN/TSAN) locally for complex C++ changes.
• TPC-DS 1T on Gluten/Bolt bundle — full suite query output matches baseline (self-tested); HashAgg micro-query qtest run 5× with same check (5/5) + perf/flame in Benchmark Results.

Breaking Changes

• No
• Yes (Description: ...)